package com.mx.fs.utils;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import com.mx.fs.FileInfo;
import com.mx.fs.FileMeta;

/**
 * 
 * @author miao
 * @since 2011-07-05
 */
public class FSUtil {

	public static final String HDFS_HOST = "hdfs://localhost:9000/";

	private static FileSystem fs = null;

	// 400k
	private static final int BUFFER_SIZE = 4096 * 100;
	private static ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);

	/**
	 * A method to get a singleton FileSystem object. Maybe use a object pool
	 * later.
	 * 
	 * @return FileSystem object
	 */
	public static FileSystem getFileSystem() {
		if (fs == null) {
			try {
				Configuration conf = new Configuration();
				fs = FileSystem.get(URI.create(HDFS_HOST), conf);
			} catch (IOException e) {
				e.printStackTrace();
			}
			return fs;
		}
		return fs;
	}

	/**
	 * @deprecated Generate a random string used as the name of a file.
	 * @return
	 */
	public static String createFileName() {
		return System.currentTimeMillis() + "";
	}

	/**
	 * @deprecated
	 * @param folder
	 * @return
	 */
	public static String createFileName(String folder) {
		if (folder.endsWith("/")) {
			return folder + System.currentTimeMillis();
		} else {
			return folder + "/" + System.currentTimeMillis();
		}

	}

	public static String createFileName(FileInfo info) {
		String folder = info.getFolder();
		if (folder.endsWith("/")) {
			return folder + info.getName() + "__" + System.currentTimeMillis();
		} else {
			return folder + "/" + info.getName() + "__"
					+ System.currentTimeMillis();
		}
	}

	public static OutputStream getFileOutStream(FileInfo info) {
		try {
			return getFileSystem().create(new Path(createFileName(info)));
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * Transfer data from SocketChannel to hdfs system.
	 * 
	 * @param sc
	 */
	public static void store(SocketChannel sc, FileInfo info) {
		WritableByteChannel wbc = Channels.newChannel(FSUtil
				.getFileOutStream(info));
		int flag;
		
		try {
			flag = sc.read(buffer);
			
			//analysis the first buffer and get file info such as filename and format.
			ByteArrayInputStream byteIn = new ByteArrayInputStream(buffer.array());
			ObjectInputStream in = new ObjectInputStream(byteIn);  
			
				FileMeta meta = (FileMeta)in.readObject();
				System.out.println(meta);
				buffer.flip();
				buffer.position(SizeCalculator.calcSize(meta));
				System.out.println(buffer.position());
			
//			buffer.flip();
//			byte[] name = new byte[100];
//			buffer.get(name);
//			System.out.println(new String(name));
		
			
			boolean firstBuffer = true;
			while (flag != -1) {
				//The first buffer contains file meta info, so shouldn't flip when using this buffer.
				if (!firstBuffer){
					buffer.flip();
				}
				wbc.write(buffer);
				buffer.compact();
				flag = sc.read(buffer);
				firstBuffer = false;
			}
			buffer.clear();
			wbc.close();

		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally {
			buffer.clear();
			try {
				wbc.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		System.out.println("upload success!!");
	}

	/**
	 * Delete all files under the given path. 
	 * Now has some problems, needs to be improved
	 * 
	 * @param path
	 * @param flag
	 *            true if you want to delete folders
	 * @return true if all files are deleted successfully
	 */
	public static boolean deleteFolder(Path path, boolean flag) {
		FileSystem fs = getFileSystem();
		try {
			FileStatus[] status = fs.listStatus(path);
			for (int i = 0; i < status.length; i++) {
				if (!flag){
					if (!status[i].isDir()){
						if (fs.delete(status[i].getPath(), false) == false) {
							System.out.println("failed to delete file: "
									+ status[i].getPath());
						}
						System.out.println("delete file:" + status[i].getPath());
					}
				}else {
					//Want to delete folders. If  folder is non empty then throw an io exception.
					//Delete all files in the folder recursionly next time.
					//I hate recursion......
					//TODO
				}
				
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return flag;
	}
}
